package ar.uba.fi.tonyvaliente.tree;

import java.util.Iterator;
import java.util.List;
import java.util.logging.Logger;

import ar.uba.fi.tonyvaliente.files.File;
import ar.uba.fi.tonyvaliente.files.FileBlock;
import ar.uba.fi.tonyvaliente.files.FileException;
import ar.uba.fi.tonyvaliente.files.FileRecord;
import ar.uba.fi.tonyvaliente.files.Persistor;
import ar.uba.fi.tonyvaliente.files.PersistorException;
import ar.uba.fi.tonyvaliente.files.RecordPosition;

/**
 * Clase que envuelve un archivo de bloques (File) donde hay un arbol B+
 * y realiza la serializacion y deserializacion de los nodos del mismo.
 * @author Diego Morello
 * @param <K> Clase a la cual deben pertenecer las claves, debe ser comparable
 * @param <V> Clase a la cual deben pertenecer los valores
 */
public class BTreeFileWrapper<K extends Comparable<K>, V> {
	
	private static Logger log = Logger.getLogger(BTreeFileWrapper.class.getName());
	
	private File file;
	private BTreeNodePersistor<K,V> nodePersistor;
	
	/**
	 * Crea una nueva instancia de la clase que apunta a un archivo de bloques
	 * donde hay un arbol B+ y sabe como serializar y deserializar las claves y valores del arbol 
	 * @param file Archivo de bloques donde se encuentra el arbol B+
	 * @param keyPersistor Persistor para serializar y deserializar las claves del arbol
	 * @param valuePersistor Persistor para serializar y deserializar los valores del arbol
	 */
	public BTreeFileWrapper(File file, Persistor<K> keyPersistor, Persistor<V> valuePersistor) {
		this.file = file;
		this.nodePersistor = new BTreeNodePersistor<K,V>(keyPersistor, valuePersistor, this); 
	}
	
	/**
	 * Cierra el archivo encapsulado
	 * @throws BTreeException si ocurre un error al cerrar el archivo
	 */
	public void close() throws BTreeException {
		try {
			file.close();
		} catch (FileException e) {
			throw new BTreeException("Error al cerrar el BTreeFileWrapper", e);
		}
	}
	
	/**
	 * Obtiene el nodo raiz del arbol. Para ello, busca en el File subyacente las propiedades
	 * rootNodeBlockNumber y rootNodeOffset, que indican el bloque y offset dentro del bloque
	 * donde se encuentra el FileRecord correspondiente al nodo raiz.
	 */
	public RecordPosition getRootNodePosition() throws BTreeException {
		String strRootBlockNumber = file.getProperty("rootNodeBlockNumber");
		String strRootOffset = file.getProperty("rootNodeOffset");
		
		if (strRootBlockNumber != null && strRootOffset != null) {
			try {
				int rootBlockNumber = Integer.parseInt(strRootBlockNumber);
				int rootOffset = Integer.parseInt(strRootOffset);
				return new RecordPosition(rootBlockNumber, rootOffset);
			} catch (NumberFormatException e) {
				throw new BTreeException("Error al obtener posicion del nodo raiz: "
						+ "valor incorrecto", e);
			}
		} else {
			throw new BTreeException("Error en el archivo de arbol: "
					+ "no se encuentra la posicion del nodo raiz");
		}
	}
	
	/**
	 * Asigna al archivo la nueva posicion del nodo raiz
	 * @param rootNode
	 * @throws BTreeException
	 */
	public void setRootNodePosition(RecordPosition pos) throws BTreeException {
		try {
			log.fine("setRootNode IN");
			log.finer("rootNodeBlockNumber: " + pos.getBlockNumber());
			log.finer("rootNodeOffset: " + pos.getOffset());
			
			file.setProperty("rootNodeBlockNumber", Integer.toString(pos.getBlockNumber()));
			file.setProperty("rootNodeOffset", Integer.toString(pos.getOffset()));
		} catch (FileException e) {
			throw new BTreeException("Error al asignar nuevo nodo raiz", e);
		}
	}
	
	/**
	 * Asigna al archivo la maxima cantidad de elementos por nodo.
	 * La misma no debe haber sido asignada previamente
	 * @param n nueva cantidad maxima de elementos por nodo
	 * @throws BTreeException Si el valor ya fue asignado, o si ocurre
	 * algun error al escribir el valor en el archivo
	 */
	public void setMaxCantidadElementosNodo(int n) throws BTreeException {
		if (n > 0) {
			if (file.getProperty("maxCantElementosNodo") == null) {
				try {
					file.setProperty("maxCantElementosNodo", Integer.toString(n));
				} catch (FileException e) {
					throw new BTreeException("Error al asignar la cantidad maxima "
							+ "de elementos por nodo", e);
				}
			} else {
				throw new BTreeException("No se puede modificar la cantidad maxima "
						+ "de elementos por nodo");
			}
		} else {
			throw new BTreeException("Maximo de elementos no valido: " + n);
		}
	}
	
	public int getMaxCantidadElementosNodo() throws BTreeException {
		String s = file.getProperty("maxCantElementosNodo");
		try {
			if (s != null) {
				return Integer.parseInt(s);
			} else {
				throw new BTreeException("La cantidad maxima de elementos por nodo " +
						"no fue asignada al archivo");
			}
		} catch (NumberFormatException e) {
			throw new BTreeException("Error en el archivo - La cantidad " +
					"maxima de elementos por nodo es incorrecta: " + s, e);
		}
	}
	
	/**
	 * Obtiene el nodo cuyo bloque y offset dentro del bloque recibe como parametro.
	 * 
	 * @param blockNumber posicion del bloque dentro del File
	 * @param offset posicion del nodo serializado dentro del bloque
	 * @return nodo recuperado
	 * @throws BTreeException si hay un error al recuperar el nodo
	 */
	public BTreeNode<K, V> getNode(int blockNumber, int offset) throws BTreeException {
		try {
			log.fine("getNode IN");
			log.finer("BlockNumber: " + blockNumber);
			log.finer("NodeOffset: " + offset);
			
			FileBlock block = file.getBlock(blockNumber);
			FileRecord record = block.getRecord(offset);
			return nodePersistor.deserialize(record.getData());
		} catch (FileException e) {
			throw new BTreeException("Error al acceder al archivo para obtener nodo", e);
		} catch (PersistorException e) {
			throw new BTreeException("Error al deserializar nodo", e);
		}
	}
	
	/**
	 * Obtiene el nodo cuya posicion recibe por parametro
	 * @param pos posicion del nodo
	 * @return nodo recuperado
	 * @throws BTreeException si hay un error al recuperar el nodo
	 */
	public BTreeNode<K, V> getNode(RecordPosition pos) throws BTreeException {
		return getNode(pos.getBlockNumber(), pos.getOffset());
	}
	
	/**
	 * Agrega un nodo al archivo.
	 * @param node nodo a agregar
	 * @return RecordPosition que contiene el numero de bloque y offset
	 * donde se escribio el nodo
	 */
	public RecordPosition appendNode(BTreeNode<K,V> node) throws BTreeException {
		try {
			
			log.fine("appendNode IN");
			try{
				log.finer("node: " + node);
			}catch(Exception e){
				
			}
			
			FileRecord r = new FileRecord(nodePersistor.serialize(node));
			RecordPosition posicion = file.addRecord(r);
			log.finer("posicion: block: " + posicion.getBlockNumber() + " offset:  " + posicion.getOffset());
			
			return posicion;
		} catch (FileException e) {
			throw new BTreeException("Error al acceder al archivo para agregar nodo", e);
		} catch (PersistorException e) {
			throw new BTreeException("Error al serializar nodo para agregar en archivo", e);
		}
	}
	
	/**
	 * Escribe un nodo en el archivo, en la posicion indicada por parametro
	 * @param node Nodo a actualizar en archivo
	 * @param pos Posicion del archivo donde escribir el nodo
	 * @return TRUE si pudo escribir el nodo, FALSE si el bloque indicado en pos estaba lleno
	 * y no pudo hacerse la actualizacion
	 * @throws BTreeException Si ocurre un error al acceder a archivo o serializar el nodo
	 */
	public boolean updateNode(BTreeNode<K,V> node, RecordPosition pos) throws BTreeException {
		try {
			
			log.fine("updateRootNode IN");
			log.finer("NodeBlockNumber: " + pos.getBlockNumber());
			log.finer("NodeOffset: " + pos.getOffset());
			try{
				log.finer("node: " + node);
			}catch(Exception e){
				
			}
			
			FileBlock block = file.getBlock(pos.getBlockNumber());
			FileRecord r = new FileRecord(nodePersistor.serialize(node));
			if (block.updateRecord(pos.getOffset(), r)) {
				file.updateBlock(pos.getBlockNumber(), block);
				return true;
			} else {
				return false;
			}
		} catch (FileException e) {
			throw new BTreeException("Error al acceder a archivo para actualizar nodo", e);
		} catch (PersistorException e) {
			throw new BTreeException("Error al serializar nodo para actualizar en disco", e);
		}
	}

	
	/**
	 * 
	 * @param node
	 * @param originalPos
	 * @return
	 * @throws BTreeException
	 */
	public RecordPosition moveNode(BTreeNode<K, V> node, RecordPosition originalPos)
			throws BTreeException {
		
		RecordPosition newPos = null;
		
		try {
			// Escribo todos los bloques cacheados
			file.flush();
			
			// Escribo el nodo en un nuevo bloque, sin actualizar referencias
			byte[] data = this.nodePersistor.serialize(node);
			newPos = file.addRecord(new FileRecord(data));
			
			// Verifico si el nodo era el ultimo del bloque
			// Si no era el ultimo, tendre que mover el ultimo al lugar que el ocupaba
			// y luego debere actualizarlo
			boolean ultimoMovido = false;
			FileRecord last = null;
			FileBlock oldBlock = file.getBlock(originalPos.getBlockNumber());
			if (originalPos.getOffset() == oldBlock.getRecordCount() - 1) {
				// Si era el ultimo, lo borro directamente
				oldBlock.eraseRecord(originalPos.getBlockNumber());
			} else {
				// Si no lo era, muevo el ultimo FileRecord a su lugar
				// (posicion inicial del nodo que quiero mover)
				last = oldBlock.getRecord(oldBlock.getRecordCount()-1);
				oldBlock.updateRecord(originalPos.getOffset(), last);
				oldBlock.eraseRecord(oldBlock.getRecordCount() - 1);
				ultimoMovido = true;
			}
			
			// Grabo nuevamente el bloque, que ya no contiene a node
			// Hasta ahora no se actualizaron referencias
			file.updateBlock(originalPos.getBlockNumber(), oldBlock);
			
			/* ----- Actualizo las referencias de node ----- */
			updateAllReferences(node, originalPos, newPos, oldBlock);
			
			/* ----- Actualizo las referencias del nodo movido ----- */
			if (ultimoMovido) {
				// Levanto el nodo desde el disco
				// y le seteo el arbol y la referencia a si mismo
				DiskBReference<K, V> thisRef = (DiskBReference<K, V>)node.getTree().createReference();
				thisRef.setRecordPosition(originalPos);
				BTreeNode<K, V> lastNode = thisRef.getNode();
				lastNode.setThisReference(thisRef);
				lastNode.setTree(node.getTree());
				
				RecordPosition originalPos2 = new RecordPosition(originalPos.getBlockNumber(), oldBlock.getRecordCount());
				RecordPosition newPos2 = originalPos;
				
				// Actualizo referencias
				updateAllReferences(lastNode, originalPos2, newPos2, oldBlock);
			}
			
		} catch (FileException e) {
			throw new BTreeException("Error al mover nodo dentro del archivo", e);
		} catch (PersistorException e) {
			throw new BTreeException("Error al mover nodo dentro del archivo", e);
		}
		
		return newPos;
	}
	
	private void updateAllReferences(BTreeNode<K, V> node,
			RecordPosition originalPos, RecordPosition newPos,
			FileBlock originalBlock) throws BTreeException {
		
		// 1. Nodo padre (salvo que node sea raiz)
		DiskBReference<K, V> parent = (DiskBReference<K, V>)node.getParent();
		if (parent != null) {
			updateParentAfterMove(parent, originalPos, newPos, originalBlock);
		} else {
			// Si es nodo raiz debo actualizar dicho valor en el archivo...
			this.setRootNodePosition(newPos);
			
			// ...y en el DiskBTree
			DiskBTree<K, V> tree = (DiskBTree<K, V>)node.getTree();
			tree.rootNodePosition = newPos;
		}
		
		// Actualizo en el nodo la referencia a si mismo con la nueva posicion
		DiskBReference<K, V> ref = (DiskBReference<K, V>)node.getThisReference();
		ref.setRecordPosition(newPos);
		
		// 2. Nodos hijos
		List<BReference<K, V>> children = node.getReferences();
		updateChildrenAfterMove(children, newPos);		
	}
	
	/**
	 * 
	 * @param parent
	 * @param originalPos
	 * @param newPos
	 * @param originalBlock
	 * @throws BTreeException
	 */
	private void updateParentAfterMove(DiskBReference<K, V> parent,
			RecordPosition originalPos, RecordPosition newPos,
			FileBlock originalBlock) throws BTreeException {
		
		boolean found = false;
		
		// Veo si el padre esta en el mismo bloque
		int nodeBlock = originalPos.getBlockNumber();
		int parentBlock = parent.getRecordPosition().getBlockNumber();
		if (nodeBlock == parentBlock) {
			// Si estan en el mismo bloque, debo ver si
			// el padre estaba en el ultimo registro
			// ya que, en tal caso, fue movido a la posicion original
			// del nodo que se reposiciono al final del archivo
			int parentOffset = parent.getRecordPosition().getOffset();
			if (parentOffset == originalBlock.getRecordCount()) {
				// El padre fue movido a la posicion original
				// del nodo reposicionado
				parent.setRecordPosition(originalPos);
			}
		}
		
		// Recorro las referencias a los hijos
		// hasta encontrar la que apunta al nodo reposicionado
		// y la actualizo con la nueva posicion del mismo
		List<BReference<K, V>> children = parent.getNode().getReferences();
		Iterator<BReference<K, V>> it = children.iterator();
		DiskBReference<K, V> child;
		while (it.hasNext() && !found) {
			child = (DiskBReference<K, V>)it.next();
			if (child.getRecordPosition().equals(originalPos)) {
				child.setRecordPosition(newPos);
				found = true;
			}
		}
		parent.updateNode();
	}
	
	/**
	 * 
	 * @param children
	 * @param newParentPos
	 * @throws BTreeException
	 */
	private void updateChildrenAfterMove(List<BReference<K, V>> children,
			RecordPosition newParentPos) throws BTreeException {
		
		// Veo si el nodo trasladado es alguno de los hijos de este
		
		DiskBReference<K, V> child;
		DiskBReference<K, V> parent;
		for (BReference<K, V> ref: children) {
			if (ref != null) {
				child = (DiskBReference<K, V>)ref;
				parent = (DiskBReference<K, V>)child.getNode().getParent();
				parent.setRecordPosition(newParentPos);
				child.updateNode();
			}
		}
	}

}
